package com.fishingwithme.domain.anybiting;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Repository
public class CircleMessageRepository {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private static final String CIRCLE_MESSAGE_KEY_PREFIX = "circle:messages:";
    private static final String CIRCLE_MESSAGE_CURSOR_PREFIX = "circle:cursor:";
    private static final Double DEFAULT_INCREMENT = 0.001;

    /**
     * 添加消息到指定 Circle
     */
    public void addMessage(String cityCode, CircleMessage message) {
        String key = CIRCLE_MESSAGE_KEY_PREFIX + cityCode;
        double score = message.getTimestamp().toInstant().toEpochMilli();
        redisTemplate.opsForZSet().add(key, message, score);

        // 限制消息数量，防止内存溢出
        Long size = redisTemplate.opsForZSet().size(key);
        if (size != null && size > 10000) {
            redisTemplate.opsForZSet().removeRange(key, 0, size - 10001);
        }
    }

    /**
     * 获取准实时消息（最近N条）
     */
    public List<CircleMessage> getQuasiRealTimeMessages(String city, int limit) {
        String key = CIRCLE_MESSAGE_KEY_PREFIX + city;
        Set<Object> messages = redisTemplate.opsForZSet()
                .reverseRange(key, 0, limit - 1);

        return messages != null ? messages.stream()
                .map(obj -> {
                    CircleMessage message = (CircleMessage) obj;
                    message.toJsonObj();
                    return message;
                })
                .collect(Collectors.toList()) : List.of();
    }

    /**
     * 获取非实时消息（分页）
     */
    public List<CircleMessage> getNonRealTimeMessages(String city, String cursor, int limit, Boolean isForward) {
        String key = CIRCLE_MESSAGE_KEY_PREFIX + city;
        double score = StringUtils.isBlank(cursor) ? 0 : Double.parseDouble(cursor);
        Set<ZSetOperations.TypedTuple<Object>> messages;
        if (isForward) {
            messages = redisTemplate.opsForZSet().rangeByScoreWithScores(key, score + DEFAULT_INCREMENT, Double.POSITIVE_INFINITY, 0, limit);
        } else {
            messages = redisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, Double.NEGATIVE_INFINITY, score - DEFAULT_INCREMENT, 0, limit);
        }
        return messages != null ? messages.stream()
                .map(tuple -> {
                    CircleMessage msg = (CircleMessage) tuple.getValue();
                    assert msg != null;
                    msg.toJsonObj();
                    return msg;
                })
                .collect(Collectors.toList()) : List.of();

    }

    /**
     * 获取最新的游标
     */
    public String getLatestCursor(String city) {
        String key = CIRCLE_MESSAGE_KEY_PREFIX + city;
        Set<ZSetOperations.TypedTuple<Object>> tuples = redisTemplate.opsForZSet()
                .reverseRangeByScoreWithScores(key, 0, Double.MAX_VALUE, 0, 1);

        return tuples != null && !tuples.isEmpty() ?
                String.valueOf(tuples.iterator().next().getScore()) : null;
    }

    /**
     * 更新用户游标
     */
    public void updateUserCursor(String city, String userId, String cursor) {
        String key = CIRCLE_MESSAGE_CURSOR_PREFIX + city + ":" + userId;
        redisTemplate.opsForValue().set(key, cursor);
    }

    /**
     * 获取用户游标
     */
    public String getUserCursor(String city, String userId) {
        String key = CIRCLE_MESSAGE_CURSOR_PREFIX + city + ":" + userId;
        return (String) redisTemplate.opsForValue().get(key);
    }
}
